package com.swak.rabbit.retry;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.rabbit.RabbitMQTemplate;
import com.swak.rabbit.RabbitMQTemplate.MessageHandler;
import com.swak.rabbit.message.Message;

/**
 * 自动重试： 需要配置全量的消费者
 * 
 * 1. 如果没有此队列的消费者则再次投递到消息队列，但是怎么知道路由等设置
 * 
 * 2. 如果本地有此队列的消费者则直接调用消费者
 * 
 * @author lifeng
 * @date 2021年3月1日 上午9:41:37
 */
public class AutoRetryConsumer extends AbstractRetryConsumer {

	protected Logger logger = LoggerFactory.getLogger(AutoRetryConsumer.class);

	@Override
	protected CompletionStage<Boolean> retry(String queue, Message message) {
		if (logger.isDebugEnabled()) {
			logger.info("Consume Queue[{}] - Message[{}] - Origin[{}] - Retry[{}] - Times[{}] Success.", queue,
					message.getId(), message.getOrigin(), message.getRetryQueue(), message.getConsumeRetrys());
		}
		MessageHandler subscriber = RabbitMQTemplate.getMessageHandler(message.getOrigin());
		if (subscriber != null) {
			return this.handleMessage(subscriber, message);
		}

		// 没有设置自动重试的，需要手动重试
		return CompletableFuture.completedFuture(Boolean.FALSE);
	}

	@SuppressWarnings({ "unchecked" })
	protected CompletionStage<Boolean> handleMessage(MessageHandler subscriber, Message message) {
		CompletableFuture<Boolean> future = new CompletableFuture<>();
		Object result = subscriber.handle(message);
		if (result != null && result instanceof CompletionStage) {
			CompletionStage<Object> resultFuture = (CompletionStage<Object>) result;
			resultFuture.whenComplete((r, e) -> {
				if (e != null) {
					future.completeExceptionally(e);
				} else if (r != null && r instanceof Boolean) {
					future.complete((Boolean) r);
				} else {
					future.complete(true);
				}
			});
		} else if (result != null && result instanceof Boolean) {
			future.complete((Boolean) result);
		} else {
			future.complete(true);
		}
		return future;
	}
}
